package main

import (
	"fmt"

	"gitee.com/xdevops/exporters/kafka-exporter/services/collect"
	"gitee.com/xdevops/exporters/kafka-exporter/services/kafka"
	"github.com/micro-plat/hydra"
	"github.com/micro-plat/hydra/conf/app"
	"github.com/micro-plat/hydra/global"
	"github.com/micro-plat/lib4go/logger"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

func init() {
	global.MQConf.PlatNameAsPrefix(false)
	//应用配置初始化后，根据配置信息创建NginxCollector实例，注册服务
	exporterApp.OnSetup(func(conf app.IAPPConf) error {

		// 从配置中心获取配置信息
		appConf, err := conf.GetServerConf().GetSubConf("app")
		if err != nil {
			return err
		}
		topic := appConf.GetString("topic")
		fields := appConf.GetStrings("fields")
		if topic == "" || len(fields) == 0 {
			return fmt.Errorf("未配置topic或field")
		}

		// 设置Nginx访问日志的文件路径
		kafka := kafka.NewKafkaService(topic)
		logger := logger.New("kafka-exporter")
		durationHours := appConf.GetInt("duration_hours", 24)

		collector := collect.NewKafkaCollector(durationHours, fields...)

		// 开始收集通过文件监控器传递的日志数据
		collector.Collect(kafka.Subscribe())

		// 创建一个新的Exporter实例,将收集器注册到Exporter中
		exporter := collect.NewExporter()
		exporter.AddCollector(collector.Collectors()...)

		// 创建一个Prometheus HTTP处理器，用于暴露指标
		h := promhttp.HandlerFor(exporter.GetGatherers(),
			promhttp.HandlerOpts{
				ErrorLog:      logger,                   // 设置错误日志记录器
				ErrorHandling: promhttp.ContinueOnError, // 设置错误处理方式为继续处理
			})

		// 注册一个HTTP服务，用于暴露指标
		exporterApp.Micro("/metrics", func(ctx hydra.IContext) interface{} {
			h.ServeHTTP(ctx.Response().GetHTTPReponse(), ctx.Request().GetHTTPRequest())
			return nil
		})

		//注册消息消费服务
		hydra.S.MQC(fmt.Sprintf("/kafka/%s/recv", topic), kafka.RecvHandle, topic)
		return nil

	}, global.API)

}
